package com.me.bigdata;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;

public class JavaWordCount {
    public static void main(String... args) {
        JavaSparkContext sc = new JavaSparkContext("local", "JavaWordCount");
        JavaRDD<String> inputData = sc.textFile(args[0]);
        JavaRDD<String> words = inputData.flatMap(line -> Arrays.asList(line.split(" ")));
        JavaPairRDD<String, Integer> result = words.filter(word -> !("".equals(word) || word.contains(","))).mapToPair(word -> new Tuple2<>(word, 1)).reduceByKey((a, b) -> a + b).mapToPair(kv -> new Tuple2<>(kv._2, kv._1)).sortByKey(false).mapToPair(kv -> new Tuple2<>(kv._2, kv._1));
        result.saveAsTextFile(args[1]);
    }
}
